package com.huitone.smspfm.core.smscallltest;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.lang.Thread;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import com.bin.common.utils.DateTimeUtils;
import com.bin.common.utils.SerialUtils;
import com.bin.common.utils.StringUtils;
import com.huitone.smspfm.context.utils.ContextUtils;
import com.huitone.smspfm.core.config.AppStaticConfigs;
import com.huitone.smspfm.core.wransms.SmsWarnRecvMessage;
import com.huitone.smspfm.core.wransms.WranSmsSocketTask;
import com.huitone.smspfm.dao.oracle.SmsCallTestDao;
import com.huitone.smspfm.dao.oracle.SmsSendRecordDao;
import com.huitone.smspfm.dao.oracle.entity.SmsCallTest;
import com.huitone.smspfm.dao.oracle.entity.SmsSendRecord;
import com.huitone.smspfm.dao.oracle.entity.WarnRecv;
import com.huitone.smspfm.logger.ExtLogger;
import com.huitone.smspfm.model.SmsEntifyForCallTest;
import com.huitone.smspfm.model.SmsEntityForMt;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;
import com.huitone.smspfm.task.GetPriListLengthTask;

public class SmsCallTestScanTask extends Thread {

	private static final ExtLogger logger = ExtLogger.create(SmsCallTestScanTask.class);
	private final static String pattern = "yyyy:MM:dd:HH:mm:ss";
	private final static String redisPattern = "yyyy:MM:dd:HH:mm:";
	private static Date queryDate = null;
	
	private boolean toSleep = true;
	
	private long sleepTime = 1000;
	
	private boolean keepGoing = true;	
	
	private long currSacnDbTime = System.currentTimeMillis();
	private long scanDbTime = 2*60*1000;
	private long currSendSmsTime = System.currentTimeMillis();
	private long sendSmsTime = 30*1000;
	private long currScanReportTime = System.currentTimeMillis();
	private long scanReportTime = 5*60*1000;
	
	private int scanPoolSize = 5;
	
	public boolean isKeepGoing() {
		return keepGoing;
	}



	public void setKeepGoing(boolean keepGoing) {
		this.keepGoing = keepGoing;
	}



	
	private ExecutorService scanPool = new ThreadPoolExecutor(scanPoolSize, 
			scanPoolSize, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(scanPoolSize * 2), new RejectedExecutionHandler() {
		
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			boolean offer = false;
			while(!executor.isShutdown() && !offer) {
				offer = executor.getQueue().offer(r);
				if (!offer) {
					try {
						TimeUnit.MILLISECONDS.sleep(10);
					} catch (InterruptedException e) {}
				}
			}
		}
	});
	
	
	
	@Override
	public void run() {
		while( keepGoing && ! this.isInterrupted()){
			try{
				long time = System.currentTimeMillis();
				//2分钟　　加载数据库
				if( time - currSacnDbTime > scanDbTime){
					currSacnDbTime = time;
					scanPool.execute(new QueryDbCallTestSms());
				}
				
				//0.5分钟一次  发送短信
				if(time - currSendSmsTime > sendSmsTime){
					currSendSmsTime = time;
					scanPool.execute(new ScanRedisCallTestSms());
				}
				
				
				// 5分钟，　检测报告
				if(time - currScanReportTime > scanReportTime){
					currScanReportTime = time;
					scanPool.execute(new ScanReportCallTestSms());
				}
				
			} catch (Exception e){e.printStackTrace(); toSleep = true;}
			
			if(toSleep){
				try {
					Thread.sleep(sleepTime);
				} catch (InterruptedException e) {}
			}
		}
	}



	private void scanReport() {
		List<SmsCallTest> dbUpdateList = new ArrayList<SmsCallTest>();
		//扫描redis中的拔测短信列表
//		logger.info("拔测短信报告扫描　"+ new Date());
		
		RedisScript<List> script = new DefaultRedisScript<List>(AppStaticConfigs.luaGetCallTestListKeys, List.class);
		List<String> luaKeys = new ArrayList<String>();
		luaKeys.add(AppStaticConfigs.isMcallTestSend);
		
		List<String> keys = AppStaticConfigs.scriptOpts3.getList(script, luaKeys, 1000+"");
		List<byte[]> saveKeys = new ArrayList<byte[]>();
		List<String> wranKeys = new ArrayList<String>();
		//查询数据库
		List<SmsSendRecord> smsRecord = new ArrayList<SmsSendRecord>();
		if(keys != null && keys.size() > 0){
			for(String key : keys){
				List<String> keyList = StringUtils.splitToList(key, ":");
				if (keyList != null && keyList.size() >= 3) {
					String seqNum = keyList.get(0);
					String phone = keyList.get(1);
					String gateLinkId = keyList.get(2);
					SmsSendRecord sms = ContextUtils.getBean2(SmsSendRecordDao.class).getObj(seqNum, phone);
					if(sms == null){
						saveKeys.add(AppStaticConfigs.stringRedisSerializer.serialize(key));
					} else {
						if(! StringUtils.equals("0", sms.getSendCode())){//失败发告警短信
							wranKeys.add(key+":"+gateLinkId);
						} 
						dbUpdateList.add(toCallTestEntity("",sms.getSendCode(),sms.getSmSeq()));
						
					}
				}
			}
		}
		
		//未出结果的键保存至缓存中
		if(saveKeys.size() > 0){
			
			byte[][] bts = new byte[saveKeys.size()][];
			for(int i  = 0; i < saveKeys.size() ; i++){
				bts[i] = saveKeys.get(i);
			}
			 AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {

				@Override
				public void excute(RedisConnection connection) {		
					
						connection.lPush(AppStaticConfigs.isMcallTestSendB,bts);
					
				}
				 
			});
		}
		
		//需要发告警短信的手机号
		if(wranKeys.size() > 0){
			
			for (String wranKey : wranKeys) {
				List<String> keyList = StringUtils.splitToList(wranKey, ":");
				if (keyList != null && keyList.size() >= 3) {
					String seqNum = keyList.get(0);
					String phone = keyList.get(1);
					String gateLink = keyList.get(2);
					// 写告警短信
					List<WarnRecv> users = AppStaticConfigs.warnRecvList;
					String mContent = AppStaticConfigs.smsTemplateMap.get(
							AppStaticConfigs.Const.SMS_TEMPLATE_TYPE_BT)
							.getSmsContent();
					
					mContent =  StringUtils.replace(mContent,"[userNum]",phone);
					mContent =  StringUtils.replace(mContent,"[gateLinkId]",gateLink);
					mContent =  StringUtils.replace(mContent,"[userNum]",seqNum);
					SmsWarnRecvMessage message = new SmsWarnRecvMessage(users,
							mContent,AppStaticConfigs.Const.SMS_TEMPLATE_TYPE_BT);

					WranSmsSocketTask.msgQueue.offer(message);
//					logger.info("发送拔测失败告警短信 ：" + message.toString());
				}
			}
		}
		
		//更新数据库记录
		if (dbUpdateList.size() > 0) {
			int cnt = ContextUtils.getBean2(SmsCallTestDao.class).updateObjAllForSeq(dbUpdateList);
//			logger.info("更新拔测短信数据库　" + cnt + "条");
		}
				
	}



	@Override
	public void destroy() {
		setKeepGoing(false);
		this.interrupt();
	}
	
	private class ScanRedisCallTestSms implements Runnable{

		@Override
		public void run() {
			scanRedisCallTestSms();
			
		}
		
	}
	
	private class ScanReportCallTestSms implements Runnable{

		@Override
		public void run() {
			scanReport();
			
		}
		
	}
	
	private class QueryDbCallTestSms implements Runnable{

		@Override
		public void run() {
			queryDb();
			
		}
		
	}
	
	private void scanRedisCallTestSms(){//扫描当前分钟内需要发送的短信
		
		List<SmsCallTest> dbUpdateList = new ArrayList<SmsCallTest>();
//		[SMCall:dateStr]
		String scanKey = AppStaticConfigs.isMCallTest + ":"+DateTimeUtils.dateToStr(new Date(), redisPattern) +"*";
		
		RedisScript<List> script = new DefaultRedisScript<List>(AppStaticConfigs.luaGetCallTestList, List.class);
		
		List<String> keys = new ArrayList<>();
		keys.add(scanKey);
		
		List<List<String>> entities = AppStaticConfigs.scriptOpts.getHashList(script, keys, 1000+"");
		
		for(int i = 0; i < entities.size(); i++){
			List<String> list = entities.get(i);
			SmsEntifyForCallTest sms = SmsEntifyForCallTest.parse(list);
			if(sms == null){
				logger.error("SmsEntifyForCallTest parse error");
				continue;
			}
			//开始组装数据发送短信
			String priName = getPriQueueName(sms.getGateLinkId());
			String svrName = AppStaticConfigs.systemSmsSendSvrNode;
			String modeName = AppStaticConfigs.appName;
			String smType = "1";
			String sendPort = AppStaticConfigs.systemSmsSendSendPort;
			
			
			String serialSegment= AppStaticConfigs.systemSmsSendCode 
						+","+ DateTimeUtils.dateToStr(new Date(),"MMddHHmmss" 
						+"," + StringUtils.toString(SerialUtils.getSerial()));
			
			String serviceCode = AppStaticConfigs.systemSmsSendServiceCode;
			List<String> entityList = new ArrayList<String>();
			entityList.add(0,svrName);				
			entityList.add(1,serviceCode);				
			entityList.add(2,smType);
			entityList.add(3,sms.getTestPhone());
			entityList.add(4,sms.getTestContent());
			entityList.add(5,"1");//短信类型
			entityList.add(6,DateTimeUtils.dateToStr(new Date(), "yyyyMMddHHmmss"));				
			entityList.add(7,modeName);			
			entityList.add(8,sendPort);				
			entityList.add(9,serialSegment);
			entityList.add(10,"");//提交时间
			entityList.add(11,"");//结果时间
			entityList.add(12,"");//旧平台编码
			entityList.add(13,serialSegment.replaceAll(",", ""));//旧平台序号
			entityList.add(14,"");//未知
			
			SmsEntityForMt smsEntity  = SmsEntityForMt.parse(entityList);
			Map<byte[],byte[]> smsMap = smsEntity.getMapB();
			String seqNum = smsEntity.getSerialno() + ":" + smsEntity.getPhoneNumber();
			AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {

				@Override
				public void excute(RedisConnection connection) {
					connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum), smsMap);//SMINFO
					//SMCache
					connection.lPush(AppStaticConfigs.stringRedisSerializer.serialize(priName),AppStaticConfigs.stringRedisSerializer.serialize(seqNum));//PriList
					GetPriListLengthTask.priListLength.put(priName, StringUtils.toLong(GetPriListLengthTask.priListLength.get(priName)) + 1);//LENGTH
					
					
					//序列号保存在缓存中
					connection.lPush(AppStaticConfigs.isMcallTestSendB, AppStaticConfigs.stringRedisSerializer.serialize(seqNum +":"+sms.getGateLinkId()));
					
					//更新数据库
					dbUpdateList.add(toCallTestEntity(sms.getRecId(),"",smsEntity.getSerialno()));
				}
				
			});

		}
		
		//更新数据库记录
		if(dbUpdateList.size() > 0){
			 int cnt = ContextUtils.getBean2( SmsCallTestDao.class).updateObjAll(dbUpdateList);
			 logger.info("更新拔测短信数据库　"+cnt +"条");
		}
		
		
		
	}



	protected SmsCallTest toCallTestEntity(String recId,String sendResult,String smSeq) {
		SmsCallTest result = new SmsCallTest();
		result.setRecId(recId);
		result.setSendResult(sendResult);
		result.setSmSeq(smSeq);
		return result;
	}



	private String getPriQueueName(String gateLinkId) {
		String result = null;
		String scanKey = AppStaticConfigs.iSMPriListNamesSet;
		
		RedisScript<List> script = new DefaultRedisScript<List>(AppStaticConfigs.luaGetPriListName, List.class);
		
		List<String> keys = new ArrayList<>();
		keys.add(scanKey);
		
		List<String> list = AppStaticConfigs.scriptOpts3.getList(script, keys, "1000");
		long length = -1L;
		if(list != null && list.size() > 0){
			for (int i = 0; i < list.size(); i++) {
				String priListName = list.get(i);
				String gateIdKey = StringUtils.splitThRIndex(priListName, ":", 1);
				if(!StringUtils.equals(gateLinkId, gateIdKey)){
					continue;
				}
				long listLength = -1L;
				if(GetPriListLengthTask.priListLength.containsKey(priListName)){
					listLength = GetPriListLengthTask.priListLength.get(priListName);
				}
				if (listLength != -1) {
					if (length == -1) {
						length = listLength;
						result = priListName;
					} else {
						if (length > listLength) {
							length = listLength;
							result = priListName;
						}
					}
				}
			}
		}
		
		if(result == null){//主动生成一个优先级队列　，　[SM:GATE_LINK_ID:999:LIST]
//			String priListNames = "SM:"+gateLinkId+":Pri:999:List";
			String priListNames = "SM:"+gateLinkId+":Cache:999:List";
			result = priListNames;
//			GetPriListLengthTask.priListLength.put(result,1L); 未添加，　不能使用
			AppStaticConfigs.pipelineOpts.save(new ABExcuteCommands() {
				@Override
				public void excute(RedisConnection connection) {
					
					connection.sAdd(AppStaticConfigs.iSMPriListNamesSetB, AppStaticConfigs.stringRedisSerializer.serialize(priListNames));
				}				
			});
			
		}
		return result;
	}



	private void queryDb(){
		List<SmsCallTest> scts = null;
		Date date = queryDate;
//		logger.info("拔测短信　task queryDb" +date);
		queryDate = new Date();
		if(date == null){
			scts = ContextUtils.getBean2(SmsCallTestDao.class).getAllObj();
		} else {
			scts = ContextUtils.getBean2(SmsCallTestDao.class).getAllObj(date);
		}
		
		Map<String,Map<byte[],byte[]>> redisMap = new HashMap<String, Map<byte[],byte[]>>();
//		if(scts != null && scts.size() > 0){
//			for(int i = 0; i < scts.size(); i++){
//				SmsCallTest sct = scts.get(i);
//				Date smsDate = sct.getSendTime();
//				String testPhone = sct.getTestPhone();
//				String testContent = sct.getTestContent();
//				String recId = sct.getRecId();
//				String gateLinkId = sct.getGateLinkId();
//				String sendTime = DateTimeUtils.dateToStr(smsDate, pattern);
//				String sendResult = sct.getSendResult();
//				String smSeq = sct.getSmSeq();
//				if(StringUtils.isEmpty(smsDate) || StringUtils.isEmpty(testPhone) || StringUtils.isEmpty(testContent) || StringUtils.isEmpty(gateLinkId)){
//					continue;
//				}
//				String dateStr = DateTimeUtils.dateToStr(smsDate, pattern);
//				
//				String key = dateStr + ":" + testPhone+":"+recId;
//				SmsEntifyForCallTest sms = new SmsEntifyForCallTest();
//				sms.setGateLinkId(gateLinkId);
//				sms.setRecId(recId);
//				sms.setSendResult(sendResult);
//				sms.setSendTime(sendTime);
//				sms.setSmSeq(smSeq);
//				sms.setTestContent(testContent);
//				sms.setTestPhone(testPhone);
//				Map<byte[],byte[]> values = sms.toMap();
//				redisMap.put(key, values);
//			}
//		}
		if(redisMap.size() > 0){
			final Map<String,Map<byte[],byte[]>> redisMapTmp = redisMap;
			AppStaticConfigs.pipelineOpts.save(new ABExcuteCommands() {

				@Override
				public void excute(RedisConnection connection) {
					
					for(String key : redisMapTmp.keySet()){
						connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.isMCallTest+":"+key),
								redisMapTmp.get(key));
					}
				}
				
			});
			
		}
	}
}
